/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog.pipe;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.changelog.PipeNormallyClosed;
import com.nhncorp.neptune.common.Constants;

public class Pipe implements Comparable<Pipe> {
  static final Log LOG = LogFactory.getLog(Pipe.class);
  public static int G_PORT = 18000;
  static final int HEART_BEAT_SIZE = 1;
  static final AtomicLong pipeCount = new AtomicLong(0);

  Object stateChangeLock = new Object();
  PipeState state;

  String tabletName;
  String pipeKey;
  Context ctx;
  String logPath;
  AtomicBoolean triggerClosing = new AtomicBoolean();
  volatile long lastTouchTime = System.currentTimeMillis();
  PipeEventHandler eventHandler;
  int pipeLifeTime;
  AtomicBoolean activated = new AtomicBoolean();
  final long pipeNo = pipeCount.getAndIncrement();

  public Pipe(SocketChannel prevChannel, int timeout) {
    this.pipeLifeTime = timeout * 1000;
    this.ctx = new Context(prevChannel);
    this.state = InitState.instance();
    activated.set(true);
  }
  
  public boolean equals(Object obj) {
    if (obj instanceof Pipe == false) {
      return false;
    }
    
    Pipe pipe = (Pipe) obj;
    return compareTo(pipe) == 0;
  }

  public int compareTo(Pipe anotherPipe) {
    return (int) (this.pipeNo - anotherPipe.pipeNo);
  }

  public void setListenPortNumber(int port) {
    this.ctx.port = port;
    G_PORT = port;
  }

  public void setEventHandler(PipeEventHandler handler) {
    this.eventHandler = handler;
  }

  public void setLogPath(String logPath) {
    this.logPath = logPath;
  }

  public void init() throws IOException {
    this.state.init(ctx);
  }

  public String getPipeKey() {
    return pipeKey;
  }

  public boolean write(SelectableChannel ch) throws IOException {
    touch();
    return ctx.isPreviousChannel(ch) ? state.writeToPrev(ctx) : state.writeToNext(ctx);
  }

  public boolean read(SelectableChannel ch) throws IOException {
    try {
      boolean isEventfromPrevious = ctx.isPreviousChannel(ch);
      //touch(isEventfromPrevious);
      touch();
      return isEventfromPrevious ? state.readFromPrev(ctx) : state.readFromNext(ctx);
    } catch (IOException e) {
      if (e instanceof PipeNormallyClosed == false) {
        if (ctx.isPreviousChannel(ch)) {
          LOG.warn("Pipe " + this + " reading from previous channel is fail", e);
        } else {
          LOG.warn("Pipe " + this + " reading from next channel is fail", e);
        }
      } else {
        LOG.debug("Pipe#" + pipeKey + " is gracefully closed");
      }
      throw e;
    }
  }

  public boolean connect(SelectableChannel ch) throws IOException {
    LOG.debug("Connect event is arrived");
    return state.connectToNext(ctx);
  }

  public void fini(int bytes) throws IOException {
    if (!activated.get()) {
      throw new IOException();
    }

    waitForRunState();

    state.fini(ctx, bytes);
  }

  public void triggerClose() {
    if (triggerClosing.getAndSet(true)) {
      // 이미 이 Pipe가 닫혀져 있는 상태
      return;
    }
    ctx.selector.wakeup();
  }

  public boolean isCloseTriggered() {
    return triggerClosing.get();
  }

  public boolean isCloseState() {
    return state.getStateName().equals("Close State");
  }

  public void close() {
    if (activated.get() == false) {
      LOG.warn("Pipe#" + pipeKey + " is already closed");
      return;
    }

//    try {
//      state.close(ctx);
//    } catch (IOException e) {
//      LOG.warn("Error in closing pipe", e);
//    }
    
    triggerClosing.set(true);

    try {
      eventHandler.pipeClosed(tabletName, pipeKey);
      eventHandler = null;
    } catch (IOException e) {
      LOG.warn("calling pipe removed event handler is fail", e);
    }

    ctx.close();
    ctx = null;

    activated.set(false);
  }

  /*
  public void rollback() throws IOException {
    if (ctx.fileChannel != null) {
      LOG.debug("rollback is called");
      ctx.fileChannel.rollback();
    }
  }
  */

  public void removeLogFile() {
    if (ctx.fileChannel != null) {
      ctx.fileChannel.removeFile();
    }
  }

  public void backupFile() throws IOException {
    if (ctx.fileChannel != null) {
      ctx.fileChannel.backup();
    }
  }

  public void restoreLastLogFile() throws IOException {
    if (ctx.fileChannel != null) {
      ctx.fileChannel.restoreLastLog();
    }
  }

  public void removeBackupFiles() {
    if (ctx.fileChannel != null) {
      ctx.fileChannel.removeBackupFiles();
    }
  }

  public long getLogFileSize() throws IOException {
    if (ctx.fileChannel == null) {
      return -1l;
    } else {
      return ctx.fileChannel.getFileSize();
    }
  }

  private void waitForRunState() throws IOException {
    synchronized (stateChangeLock) {
      while (state != RunState.instance()) {
        try {
          LOG.debug("wait for state changing");
          stateChangeLock.wait();
        } catch (InterruptedException e) {
          LOG.warn("awake from waiting due to interrupting", e);
        }

        if (!activated.get()) {
          LOG.debug("awake from waiting due to pipe closed");
          throw new IOException("Pipe has been closed");
        }
      }
    }
  }

  public void changeState(PipeState newState) throws IOException {
    synchronized (stateChangeLock) {
//      this.state.shutdown(ctx);
      newState.init(ctx);
      this.state = newState;

      stateChangeLock.notifyAll();
    }
  }

  public boolean isExpired(long processTime) {
    if (ctx.checkValidity.get()) {
      return System.currentTimeMillis() - lastTouchTime > pipeLifeTime + processTime;
    }
    
    return false;
  }

  public boolean isActivated() {
    return this.activated.get();
  }

//  public void touch(boolean fromPrevious) {
  public void touch() {
    lastTouchTime = System.currentTimeMillis();
    // Pipe가 touch(false)로 호출 되었다는 의미는 이 pipe에 최근에 쓰여진 데이터가 성공적으로 저장되었기 때문에
    // Tablet서버가 다음 동작을 했다는 것을 의미하기 때문에 rollback index를 file size로 초기화시킨다.
    // fromPrevious가 false일 경우는 tablet server로 부터 발생한 이벤트가 아니기 때문에 file을
    // 건드리지 않는다.
    /*
    if (fromPrevious && ctx != null && ctx.fileChannel != null) {
      LOG.debug("reset rollback index in touch");
      ctx.fileChannel.resetRollbackIndex();
    }
    */
  }

  public String toString() {
    return "Pipe#" + pipeKey
      + ", state : " + ((state == null) ? "NULL" : state.getStateName())
      + " inReceivingData : " + ctx.inReceivingData.get()
      + " inReturning : " + ctx.inReturning.get()
      + " from [" + ((ctx.prevChannel == null) ? "NULL" : ctx.prevChannel.socket().getRemoteSocketAddress()) 
      + "] to [" + ((ctx.nextChannel == null) ? "NULL" : ctx.nextChannel.socket().getRemoteSocketAddress())
      + "]";
  }

  public void register(Selector selector) {
    if (ctx.selector != null) {
      ctx.clear();
    }

    ctx.selector = selector;
  }

  class Context {
    Selector selector;
    SocketChannel prevChannel;
    SocketChannel nextChannel;
    boolean isLastPipe = false;

    Message msg = new Message();
    Bulk bulk;
    ByteBuffer heartBeatBuf = ByteBuffer.allocateDirect(HEART_BEAT_SIZE);
    ChangeLogFileChannel fileChannel;

    boolean selectingWrite;
    Object numReadingBytesLock = new Object();
    int numReadingBytes;;
    int port;
    
    AtomicBoolean inReceivingData = new AtomicBoolean(false);
    AtomicBoolean inReturning = new AtomicBoolean(false);
    AtomicBoolean checkValidity = new AtomicBoolean(true);
    long fileSize;

    public Context(SocketChannel prevChannel) {
      this.prevChannel = prevChannel;
    }

    public void close() {
      clear();
      
      msg = null;
      if (bulk != null) {
        bulk.close();
      }
      bulk = null;
      prevChannel = null;
      nextChannel = null;
      fileChannel = null;
    }

    public void clear() {
      msg.clear();
      if (bulk != null) {
        bulk.clear();
      }

      try {
        prevChannel.close();
      } catch (IOException e) {
        LOG.warn("closing previous channel is fail", e);
      }

      try {
        if (nextChannel != null)
          nextChannel.close();
      } catch (IOException e) {
        LOG.warn("closing next channel is fail", e);
      }

      if (fileChannel != null) {
        try {
          ChangeLogFileChannel.closeChannel(fileChannel);
          fileChannel = null;
        } catch (IOException e) {
          LOG.warn("closing file channel is fail", e);
        }
      }

      numReadingBytes = 0;
      isLastPipe = false;
      selector = null;
    }

    boolean isPreviousChannel(SelectableChannel ch) {
      return ch == prevChannel;
    }

    long prepare(PipeConnectionInfo pipeInfo, int pipePosition) throws IOException {
      tabletName = pipeInfo.getTabletName();
      pipeKey = pipeInfo.getPipeKey();
      
      fileChannel = ChangeLogFileChannel.createChannel(Pipe.this.tabletName, Pipe.this.logPath,
          Constants.PIPE_CL_FILE_NAME + ctx.port);
      eventHandler.pipeEstablished(tabletName, pipeKey, Pipe.this);
      fileSize = fileChannel.getFileSize(); 
      LOG.debug("New pipe# " + pipeKey + ", pipe pos : " + pipePosition + ", fileSize : " + fileSize);
      return fileSize; 
    }

    void cancelPreparation() {
      try {
        LOG.info("Cancel pipe [" + pipeKey + "] preparation");
        ChangeLogFileChannel.closeChannel(fileChannel);
        fileChannel = null;
        tabletName = null;
      } catch (IOException e) {
        LOG.info(e);
      }
    }

    void registerToSelect(SelectableChannel ch, int opCode) throws IOException {
      SelectionKey key = ch.keyFor(selector);

      if (key == null) {
        key = ch.register(selector, opCode);
        key.attach(Pipe.this);
      } else if (key.isValid() == false) {
        key = ch.register(selector, opCode);
        key.attach(Pipe.this);
      } else {
        key.interestOps(key.interestOps() | opCode);
      }
    }

    public void deregisterFromSelect(SocketChannel ch, int opCode) throws IOException {
      SelectionKey key = null;

      if ((key = ch.keyFor(selector)) != null) {
        key.interestOps(key.interestOps() & ~opCode);
      }
    }
    
    public void deregisterFromSelectAll(SocketChannel ch) {
      SelectionKey key = null;

      if ((key = ch.keyFor(selector)) != null) {
        key.interestOps(0);
        key.cancel();
      }
    }

    public void changePipeState(PipeState state) throws IOException {
      Pipe.this.changeState(state);
    }

    public void waitForReadingDone(int bytes) throws IOException {
      int wakeupCount = 0;

      long elapsed = 0;
      synchronized (numReadingBytesLock) {
        while (numReadingBytes < bytes) {
          try {
            if (wakeupCount > 0) {
              LOG.debug("waiting for " + bytes + " bytes read, current : " + numReadingBytes);
            }
            long before = System.nanoTime();
            numReadingBytesLock.wait(1000);
            elapsed += (System.nanoTime() - before);
            wakeupCount++;
          } catch (InterruptedException e) {
            LOG.debug("awake from waiting due to interrupting");
          }

          if (!Pipe.this.activated.get()) {
            LOG.debug("awake from waiting due to pipe closed");
            throw new IOException("Pipe has been closed");
          }

          if (wakeupCount > 1 && numReadingBytes >= bytes) {
            LOG.debug("finish waiting and clear numReading numReadingBytes : " + numReadingBytes
                + ", expected bytes : " + bytes + ", time : " + (elapsed / 1000000.0) + " ms");
            break;
          }
        }
        numReadingBytes = 0;
      }
    }

    public void setReadingNum(int numRead) {
      try {
      } catch (Exception e) {
        e.printStackTrace();
      }
      synchronized (numReadingBytesLock) {
        numReadingBytes += numRead;
        // LOG.debug("total numRead : " + numReadingBytes);
        numReadingBytesLock.notify();
      }
    }

    public String getMyInetAddress() {
      try {
        return InetAddress.getLocalHost().getHostAddress() + ":" + port;
      } catch (UnknownHostException e) {
        LOG.warn("Cannot reteive the local ip address");
      }

      return "127.0.0.1:" + port;
    }
  }
}
